package interceptor;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

public class ShuJiaInterceptor implements Interceptor {
    @Override
    public void initialize() {
        //可以初始化一些外部的链接对象 mysql redis ck ...
    }

    @Override
    public Event intercept(Event event) {
        //判断监控到的数据内容中是否有shujia这样的单词
        //如果有，就在这个event的header中添加一个键值对 type:sj
        //如果没有，就在这个event的header中添加一个键值对 type:nsj
        //获取监控到的数据内容
        String info = new String(event.getBody(), 0, event.getBody().length);

        //获取event的header数据
        Map<String, String> headers = event.getHeaders();
        if(info.contains("shujia")){
            headers.put("type","sj");
        }else {
            headers.put("type","nsj");
        }

        event.setHeaders(headers);
        return event;
    }

    @Override
    public List<Event> intercept(List<Event> list) {
        ArrayList<Event> events = new ArrayList<>();
        for (Event event : list) {
            events.add(intercept(event));
        }
        return events;
    }

    @Override
    public void close() {
      //释放在initialize创建的链接对象
    }

    public static class MyBuilder implements Builder{

        @Override
        public Interceptor build() {
            return new ShuJiaInterceptor();
        }

        @Override
        public void configure(Context context) {
            //...
        }
    }

}
